package org.dt.common.sequence;

import lombok.extern.slf4j.Slf4j;

import org.dt.common.sequence.config.SeqHolderSpec;
import org.springframework.scheduling.annotation.Async;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 【Serial Number Generator】. use {@link ConcurrentLinkedDeque} as the local cache
 *
 * <p>creation time: 2018/11/29 14:29</p>
 *
 * @author yujiadong
 * @version 0.1
 * @since 0.1
 * Changes:
 * 1) 支持同步+异步补充号段缓存
 * <p>
 * Issues:
 * 1.16: 高并发调用 applyFor(Long)时，sequencePool 瞬间poll()量会迅速消耗缓存的 sequence，造成缓存穿透，引起热点行问题
 * 1.16: supplySequenceToCache()没有控制并发量，容易在突然到来的高峰并发情况时，发起过多的数据库请求
 * 1.16: 一次 applyFor(Long)会触发sequencePool的一次poll()和offer()，可优化
 * 1.16: {@link ConcurrentLinkedDeque#size()}时间复杂度为O(N)，被Sonar认为是Block问题，因此使用额外的{@link GreedyPollSequenceHolder#cachedWidth}记录
 **/
@Slf4j
@Deprecated
public class GreedyPollSequenceHolder extends AbstractSequenceHolder {

    private AtomicLong                      cachedWidth  = new AtomicLong();
    private ConcurrentLinkedDeque<Sequence> sequencePool = new ConcurrentLinkedDeque<>();

    public GreedyPollSequenceHolder(String biz, SeqHolderSpec config, BaseDelegator stepGenerator, Executor taskExecutor) {
        super(biz, config, stepGenerator, taskExecutor);
    }

    @Override
    public List<Sequence> doApplyFor(Long rangeWidthNeeded) {
        AtomicLong                      range = this.cachedWidth;
        ConcurrentLinkedDeque<Sequence> queue = sequencePool;

        boolean needReplenish = range.get() < capacity * loadFactor;
        if (needReplenish) {
            taskExecutor.execute(this::supplySequenceToCache);
        }

        LinkedList<Sequence> list        = new LinkedList<>();
        long                 widthNeeded = rangeWidthNeeded;
        Sequence             sequence;
        while (widthNeeded != 0) {
            sequence = queue.poll();

            if (sequence != null) {
                long     seqWidth    = sequence.width();
                Sequence seqToReturn = sequence;
                range.addAndGet(-seqWidth);

                if (seqWidth > widthNeeded) {
                    Sequence seqToRuminate = new Sequence(sequence.getLowerBound() + widthNeeded, sequence.getUpperBound());
                    queue.offerFirst(seqToRuminate);
                    range.addAndGet(seqToRuminate.width());
                    seqToReturn = new Sequence(sequence.getLowerBound(), sequence.getLowerBound() + widthNeeded - 1);
                }

                widthNeeded -= seqToReturn.width();
                list.add(seqToReturn);
            } else {
                log.warn("get null when applying for sequence '{}' from cache, which may cause performance loss", biz);
            }

            if (!queue.isEmpty()) {
                continue;
            }

            Sequence fromDb = delegator.applyForOneStep(biz, capacity);
            queue.offer(fromDb);
        }
        return list;
    }

    @Override
    public void init() {
        supplySequenceToCache();
        delegator.init(biz);
    }

    @Async
    void supplySequenceToCache() {
        Sequence sequence = delegator.applyForOneStep(biz, capacity);
        if (sequence != null && sequencePool.offer(sequence)) {
            cachedWidth.addAndGet(sequence.width());
        }
    }


}

